#include <stdio.h>
#include <stdint.h>
#include <string.h>
#include <unistd.h>

#include "ohos_init.h"
#include "cmsis_os2.h"
#include <unistd.h>
#include "hi_wifi_api.h"

#include "lwip/ip_addr.h"
#include "lwip/netifapi.h"
#include "lwip/sockets.h"
#include "MQTTPacket.h"
#include "transport.h"

#include "mqtt_fun.h"

#define EVENT_STACK_SIZE    4096
#define LOOP_INTERVAL       40000
#define SUB_TOPIC_MAX       50

typedef enum 
{    
    MQTT_UNINIT = 0,
    MQTT_INIT       = 1,
    MQTT_CONNECT    = 2,
    MQTT_SUBTOPIC    = 3,
    MQTT_RECEIVE    = 4,
}MQTT_STATUS;

static volatile MQTT_STATUS gMQTTStatus = MQTT_UNINIT;
static volatile int gSubTopicCount = 0;
static volatile int gSubTopicAck = 0;
static volatile int gUnSubTopicAck = 0;
static volatile int gPingStatus = 0;

int mqttsock = 0;


typedef struct
{
    char* sub_topic;
    PMQTTMessageCallback callback;
} Sub_Topic_Info;

static volatile Sub_Topic_Info gSub_Topic_Info[SUB_TOPIC_MAX] = {0};
static volatile PMQTTStatusCallback ConnectStatusCallback;

static void MQTTMessageHandler(void)
{
    int i = 0;
	unsigned char buf[200];
	int buflen = sizeof(buf);
    int read_ack = MQTTPacket_read(buf, buflen, transport_getdata);
    if (read_ack == PUBLISH)
    {
        unsigned char dup;
        int qos;
        unsigned char retained;
        unsigned short msgid;
        int payloadlen_in;
        unsigned char* payload_in;
        int rc;
        MQTTString receivedTopic;
        rc = MQTTDeserialize_publish(&dup, &qos, &retained, &msgid, &receivedTopic,
                &payload_in, &payloadlen_in, buf, buflen);
        printf("message arrived %.*s\n", payloadlen_in, payload_in);
        if (rc)
        {
            char topic[100];
            char message[512];
            memset(topic,0,sizeof(topic));
            memset(message,0,sizeof(message));
            memcpy(topic, receivedTopic.lenstring.data, receivedTopic.lenstring.len);
            memcpy(message, payload_in, payloadlen_in);
            printf(" topic[%s] payload[%s]\r\n", topic, message);
            for(i=0; i<gSubTopicCount; i++)
            {
                if(strcmp(gSub_Topic_Info[i].sub_topic, topic) == 0 )
                {
                    gSub_Topic_Info[i].callback(topic, message);
                }
            }
        }
    } else if (read_ack == PUBACK)
    {
        printf("Publish Topic ok\n");

    } else if (read_ack == SUBACK)
    {
		unsigned short submsgid;
		int subcount;
		int granted_qos;

		gSubTopicAck = MQTTDeserialize_suback(&submsgid, 1, &subcount, &granted_qos, buf, buflen);
		if (granted_qos != 0)
		{
			printf("granted qos != 0, %d\n", granted_qos);
            gSubTopicAck = -1;
		}
	} else if (read_ack == UNSUBACK)
	{
		unsigned short unsubmsgid;
		gUnSubTopicAck = MQTTDeserialize_unsuback(&unsubmsgid, buf, buflen);
	} else if (read_ack == PINGRESP)
    {
        printf("Pong\n");
        gPingStatus = 1;
    }
}

//MQTT初始化
int MQTT_Connect(void)
{
	MQTTPacket_connectData data = MQTTPacket_connectData_initializer;
	int rc = 0;
	unsigned char buf[200];
	int buflen = sizeof(buf);
	int len = 0;

	mqttsock = transport_open(_host, _port);
	if(mqttsock < 0)
		return MQTT_ACK_SOCKET_ERR;

	printf("Sending to hostname %s port %d\n", _host, _port);

	data.clientID.cstring = _client_id;
	data.keepAliveInterval = _interval;
	data.cleansession = 1;
	data.username.cstring = _username;
	data.password.cstring = _password;

	len = MQTTSerialize_connect(buf, buflen, &data);
	rc = transport_sendPacketBuffer(mqttsock, buf, len);

	/* wait for connack */
	if (MQTTPacket_read(buf, buflen, transport_getdata) == CONNACK)
	{
		unsigned char sessionPresent, connack_rc;

		if (MQTTDeserialize_connack(&sessionPresent, &connack_rc, buf, buflen) != 1 || connack_rc != 0)
		{
			printf("Unable to connect, return code %d\n", connack_rc);
			return MQTT_ACK_CONNET_BADACK;
		}
	}
	else
    {
		return MQTT_ACK_NOACK;
    }

	printf("connect ret is %d\r\n",rc);
    gMQTTStatus = MQTT_CONNECT;
	return rc;
}

void MQTT_Disconnect(void)
{
    gMQTTStatus = MQTT_INIT;
    transport_close(mqttsock);
}

int MQTT_Subscribe_Topic(char* sub_topic, PMQTTMessageCallback callback)
{
    MQTTString topicString = MQTTString_initializer;
	int req_qos = 0;
	unsigned char buf[200];
	int buflen = sizeof(buf);	
	int len = 0;
    int msgid = 1;
	int rc = 0;
    printf("MQTT_Subscribe_Topic start\n");
    if(gSubTopicCount <= SUB_TOPIC_MAX){
        gSub_Topic_Info[gSubTopicCount].sub_topic = sub_topic;
        gSub_Topic_Info[gSubTopicCount].callback = callback;
        gSubTopicCount++;
    }
    else{
		printf("SUB_TOPIC_MAX = %d gSubTopicCount = %d\n", SUB_TOPIC_MAX,gSubTopicCount);
        return -1;
    }

    topicString.cstring = sub_topic;
	len = MQTTSerialize_subscribe(buf, buflen, 0, msgid, 1, &topicString, &req_qos);
	rc = transport_sendPacketBuffer(mqttsock, buf, len);
    MQTTMessageHandler();
    rc = rc;
    gMQTTStatus = MQTT_SUBTOPIC;
    printf("MQTT_Subscribe_Topic ok\n");
	return gSubTopicAck; 
}

int MQTT_UnSubscribe_Topic(char* unsub_topic)
{
    MQTTString topicString = MQTTString_initializer;
	unsigned char buf[200];
	int buflen = sizeof(buf);	
	int len = 0;
    int msgid = 1;
    gUnSubTopicAck = -1;
    printf("MQTT_UnSubscribe_Topic start\n");
    for(int i=0; i<gSubTopicCount; i++)
    {
        if(gSub_Topic_Info[i].sub_topic == unsub_topic)
        {
            gSub_Topic_Info[i].sub_topic = 0;
            gSub_Topic_Info[i].callback = 0;
            gSubTopicCount--;
        }
    }

    if(gSubTopicCount == 0){
        printf("MQTT_UnSubscribe_Topic start gSubTopicCount == 0 gMQTTStatus = MQTT_CONNECT;\n");
        gMQTTStatus = MQTT_CONNECT;
    }

    topicString.cstring = unsub_topic;
	len = MQTTSerialize_unsubscribe(buf, buflen, 0, msgid, 1, &topicString);
	gUnSubTopicAck = transport_sendPacketBuffer(mqttsock, buf, len);
    MQTTMessageHandler();
    return gUnSubTopicAck;
}

void MQTT_Public_Topic(char* pub_topic, char* message)
{
	MQTTString topicString = MQTTString_initializer;
	unsigned char buf[200];
	int buflen = sizeof(buf);		
	int len = 0;
	int rc = 0;		
	topicString.cstring = pub_topic;
	len = MQTTSerialize_publish(buf, buflen, 0, 0, 0, 0, topicString, (unsigned char*)message, strlen(message));
    rc = transport_sendPacketBuffer(mqttsock, buf, len);
    rc = rc;
}

void MQTT_PingREQ_Callback(PMQTTStatusCallback statusCallback)
{
	ConnectStatusCallback = statusCallback;
}

void MQTT_Check_Connect_Status(void)
{
	unsigned char buf[200];
	int buflen = sizeof(buf);		
	int len = 0;
    gPingStatus = 0;
	len = MQTTSerialize_pingreq(buf, buflen);
	transport_sendPacketBuffer(mqttsock, buf, len);
	printf("Ping...\n");
    MQTTMessageHandler();
    if(gPingStatus){
        printf("Pong...\n");
    } else {
        printf("OOPS...\n");
    }
    ConnectStatusCallback(gPingStatus);
}


static void* MQTT_Task(const char* arg)
{
    (void)arg;
    int loop_count = 0;
    while(gMQTTStatus >= MQTT_INIT)
    {
        if(gMQTTStatus >= MQTT_SUBTOPIC)
        {
            //printf("loop_count =%d...",loop_count);
            if(loop_count  >= _checkstatus){
                MQTT_Check_Connect_Status();
                loop_count = 0;
            } else {
                MQTTMessageHandler();
            }
            loop_count ++;
        } else {
            loop_count = 0;
        }
        usleep(LOOP_INTERVAL);
    }
    return NULL;
}

int MQTT_Init(void)
{
    int ret = -1;
    if( gMQTTStatus == MQTT_UNINIT)
    {
        osThreadAttr_t attr = {0};

        attr.name = "MQTT_Task";
        attr.attr_bits = 0U;
        attr.cb_mem = NULL;
        attr.cb_size = 0U;
        attr.stack_mem = NULL;
        attr.stack_size = EVENT_STACK_SIZE;
        attr.priority = osPriorityNormal;
        gMQTTStatus = MQTT_INIT;
        if(osThreadNew((osThreadFunc_t)MQTT_Task, NULL, &attr) == NULL){
            printf("Falied to create MQTT_Task!\n");
            gMQTTStatus = MQTT_UNINIT;
        } else {
            printf("success to create MQTT_Task!\n");
            ret = 0;
        }
    }

	printf("MQTT_Init is %d\r\n",ret);
    return ret;
}

void MQTT_Close(void){
    gMQTTStatus = MQTT_UNINIT;
}